package ai.people.core.standard.fill.service;

import ai.people.core.common.util.ApplicationContextUtil;
import ai.people.core.standard.fill.entity.ExtendConfig;
import ai.people.core.standard.fill.entity.Msg;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;

/**
 * 新账号推送到采集端，用于补全用户信息
 *
 * @author zhenglin
 * @since 2022/3/16
 */
public interface SendNewAccountToTopicService{
    Logger LOGGER = LoggerFactory.getLogger(SendNewAccountToTopicService.class);


    /**
     * 发送新账户数据
     *
     * @param topic         主题
     * @param list          消息列表
     * @param kafkaTemplate 卡夫卡模板
     */
    default void sendNewAccountData(String topic, List<Msg> list, KafkaTemplate kafkaTemplate){
        sendNewAccountData(topic,list,kafkaTemplate,new ExtendConfig());
    }

    /**
     * 发送新账户数据
     *
     * @param topic         主题
     * @param msg           消息
     * @param kafkaTemplate 卡夫卡模板
     */
    default void sendNewAccountData(String topic, Msg msg,KafkaTemplate kafkaTemplate){
        sendNewAccountData(topic,msg,kafkaTemplate,new ExtendConfig());
    }

    /**
     * 发送新账户数据
     *
     * @param topic         主题
     * @param list          消息集合
     * @param kafkaTemplate kafkaTemplate
     */
    default void sendNewAccountData(String topic, List<Msg> list, KafkaTemplate kafkaTemplate, ExtendConfig config) {
        if (CollectionUtils.isEmpty(list)){
            return;
        }
        ExecutorService executorService = ApplicationContextUtil.getBean(ExecutorService.class, "defaultThreadPoolExecutor");
        executorService.execute(() -> {
            List<List<Msg>> partition = Lists.partition(list, config.getPart());
            for (List<Msg> msgs : partition) {
                String msg = JSONObject.toJSONString(msgs);
                kafkaTemplate.send(topic, msg);
                LOGGER.info("推送topic:{} 数据条数:{}", topic, msgs.size());
            }
        });
    }



    /**
     * 发送新账户数据
     *
     * @param topic         主题
     * @param msg           消息
     * @param kafkaTemplate kafkaTemplate
     */
    default void sendNewAccountData(String topic, Msg msg,KafkaTemplate kafkaTemplate,ExtendConfig config){
        sendNewAccountData(topic, Collections.singletonList(msg),kafkaTemplate,config);
    }
}
